package dmp.beans.sparksql

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * author:CN.CDG
  * Date:2019/2/14
  * Time:16:23
  **/
object Net_Mysql {
  def main(args: Array[String]): Unit = {
    if(args.length!=1){
      println(
        """
          |参数不合法
          |inputpath
          |程序退出
        """.stripMargin)
      sys.exit()
    }
    val Array(inputPath)=args
    val conf=new SparkConf().setMaster("local[*]").setAppName("Operator_Mysql")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")//rdd序列化到磁盘，用于worker和worker之间的传输
    //"spark.serializer","org.apache.spark.serializer.KryoSerializer
    val spark=SparkSession.builder().config(conf).getOrCreate()
    //val sQLContext =new SQLContext(spark)

    //读取parquet文件
    val parquetData: DataFrame = spark.read.parquet(inputPath)
    //拿到数据后注册一张临时表
    parquetData.createOrReplaceTempView("log")
    val result: DataFrame = spark.sql(
      """
        |select networkmannername,
        |sum(case when requestmode = 1 and processnode >=1 then 1 else 0 end) as `原始请求`,
        |sum(case when requestmode = 1 and processnode >=2 then 1 else 0 end) as `有效请求`,
        |sum(case when requestmode = 1 and processnode = 3 then 1 else 0 end) as `广告请求`,
        |sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end) as `参与竞价数`,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) as `竞价成功数`,
        |sum(case when iseffective = 1 and isbilling = 1 and isbid = 1 then 1 else 0 end)/sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 and adorderid <> 0 then 1 else 0 end) as `竞价成功率`,
        |sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end) as `展示量`,
        |sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as `点击量`,
        |sum(case when requestmode = 2 and iseffective = 1 then 1 else 0 end)/sum(case when requestmode = 3 and iseffective = 1 then 1 else 0 end) as `点击率`,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then winprice else 0 end)/1000 as `广告成本`,
        |sum(case when iseffective = 1 and isbilling = 1 and iswin = 1 then adpayment else 0 end)/1000 as `广告消费`
        |from log
        |group by networkmannername order by networkmannername
      """.stripMargin
    )
    val load=ConfigFactory.load()
    val props=new Properties()
    props.setProperty("user",load.getString("jdbc.user"))
    props.setProperty("password",load.getString("jdbc.password"))
    result.write.jdbc(load.getString("jdbc.url"),load.getString("jdbc.tablename4"),props)
    spark.stop()
  }
}
